package com.aqielife.demo.service.impl;

import com.aqielife.demo.entity.User;
import com.aqielife.demo.service.RabbitmqService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import static com.aqielife.demo.MQConstant.*;

/**
 * @author aqie
 * @date 2022/02/06 16:52
 * @desc
 */
@Slf4j
@Service
public class RabbitmqServiceImpl implements RabbitmqService {
  @Autowired
  private RabbitTemplate rabbitTemplate;

  /**
   * confirm机制
   */
  private final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
      if (b) {
        log.info("confirm 消息确认成功..." + correlationData.getId());
      } else {
        log.info("confirm 消息确认失败..." + correlationData.getId() + " cause: " + s);
      }
    }
  };

  /**
   * return 机制
   */
  private final RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback() {
    /**
     *  return 的回调方法（找不到路由才会触发）
     * @param message 消息的相关信息
     * @param i 错误状态码
     * @param s 错误状态码对应的文本信息
     * @param s1 交换机的名字
     * @param s2 路由的key
     */
    @Override
    public void returnedMessage(Message message, int i, String s, String s1, String s2) {
      System.out.println(message);
      System.out.println(new String(message.getBody()));
      System.out.println(i);
      System.out.println(s);
      System.out.println(s1);
      System.out.println(s2);
    }
  };


  @Override
  public void sendWork() {
    for (int i = 0; i < 2; i++) {
      rabbitTemplate.convertAndSend(QUEUE_WORK, new User(i, "km", "km123"),  new CorrelationData("" + System.currentTimeMillis()));
      rabbitTemplate.setConfirmCallback(confirmCallback);
    }
  }

  @Override
  public void sendPublish() {
    for (int i = 0; i < 5; i++) {
      // 只有确定消费者接收到消息，才会发送下一条信息
      // rabbitTemplate.convertSendAndReceive("exchange_fanout", "", "测试发布订阅模型：" + i);
      rabbitTemplate.convertAndSend(EXCHANGE_FANOUT, "", "测试发布订阅模型：" + i);
    }
  }

  @Override
  public void sendTopic() {
    for (int i = 0; i < 3; i++) {
      if (i % 2 == 0) {
        rabbitTemplate.convertSendAndReceive(EXCHANGE_TOPIC, "topic.km.topic", "测试发布订阅模型：" + i);
      } else {
        rabbitTemplate.setReturnCallback(returnCallback);
        rabbitTemplate.convertSendAndReceive(EXCHANGE_TOPIC, "topic.km", "测试发布订阅模型：" + i);
      }
    }
  }
}
